package com.gonsin.easycount.service.impl;

import com.gonsin.easycount.bus.EasyBus;
import com.gonsin.easycount.enums.Caches;
import com.gonsin.easycount.plugin.OuterCache;
import com.gonsin.easycount.plugin.OuterLock;
import com.gonsin.easycount.model.UserInfoOption;
import com.gonsin.easycount.plugin.impl.TopicNotifier;
import com.gonsin.easycount.service.ICountService;
import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.stream.Collectors;

import static com.google.common.collect.Iterables.isEmpty;


@Slf4j
public class CountServiceImpl implements ICountService {

    private OuterCache cache;
    private OuterLock lock;

    private TopicNotifier topicNotifier;

    public CountServiceImpl(OuterLock lock, OuterCache cache) {
        this.cache = cache;
        this.lock = lock;
        this.topicNotifier = new TopicNotifier(lock, cache);
    }

    @Override
    public void init(String topic, List<UserInfoOption> allOptions, Object holder) {

        lock.writeLock(topic);
        try {
            // 清空redis上的缓存
//            String hashName = topic + "@";


            Map<String, UserInfoOption> lastOptionMap = cache.getHash(Caches.OPTION_HASH_MAP(topic), UserInfoOption.class);
            Set<String> options = new HashSet<>();
            for(UserInfoOption ops : lastOptionMap.values()){
                options.addAll(Arrays.asList(ops.getOption()));
            }
//            map.put(userInfo, option);

            // 上一个选票的删除掉
            if(!isEmpty(options)){
                for(String op : options){
//                    String setName = topic + "-" + op + "-" + "increase";
//                    redisTemplate.opsForSet().remove(setName, userInfo);
                    cache.cleanSet(Caches.USER_OPTION_SET(topic, op));
                }
            }


            Map<String, Set<UserInfoOption>> allOptionsMap = new HashMap<>();
            allOptions.forEach(u -> {
                for(String op : u.getOption()){
                    Set<UserInfoOption> uuSet = allOptionsMap.computeIfAbsent(op, uu -> new HashSet<>());
                    uuSet.add(u);
                }
            });
            Set<String> setNames = allOptionsMap
                    .keySet()
                    .stream()
                    .map(k -> Caches.USER_OPTION_SET(topic, k))
                    .collect(Collectors.toSet());
            cache.cleanSet(setNames);


            cache.cleanHash(Caches.OPTION_HASH_MAP(topic));
            cache.cleanValue(Caches.HOLDER(topic));
            cache.cleanValue(Caches.HOLDER_TYPE(topic));
//            cache.cleanValue(Caches.EXTRA(topic));
//            cache.cleanValue(Caches.EXTRA_TYPE(topic));
            cache.cleanList(Caches.NEW_USER_LIST(topic));




            if(holder != null){
                cache.put(Caches.HOLDER(topic), holder);
                cache.put(Caches.HOLDER_TYPE(topic), holder.getClass().getName());
            }


            // 添加新的值
//            BoundHashOperations<String, Object, Object> map = redisTemplate.boundHashOps(hashName);
            for(UserInfoOption u : allOptions){
                cache.putHash(Caches.OPTION_HASH_MAP(topic), u.getUserInfo(), u);
//                map.put(u.getUserInfo(), u.getOption());

                // 统计选项人数
                for(String op : u.getOption()){
                    cache.addSet(Caches.USER_OPTION_SET(topic, op), u.getUserInfo());
                }

//                cache.addSet(topic + "@new", u);

                // 推送到定时器，由定时器自行决定什么时候产生回调。
//                topicNotifier.push(topic, u);
            }


        } catch (Exception ex) {
            log.error(ex.toString(), ex);
        } finally {
            lock.writeUnlock(topic);
        }

    }

    @Override
    public void increase(String topic, String userInfo, String[] option, Object extraData) {

        lock.writeLock(topic);
        try {
//            String hashName = topic + "@";
//            BoundHashOperations<String, Object, Object> map = redisTemplate.boundHashOps(hashName);

//            String[] lastOptions = (String[]) map.get(userInfo);
            UserInfoOption lastOptions = cache.getHash(Caches.OPTION_HASH_MAP(topic), userInfo, UserInfoOption.class);
            UserInfoOption uio = new UserInfoOption(userInfo, option, extraData);
            cache.putHash(Caches.OPTION_HASH_MAP(topic), userInfo, uio);
//            map.put(userInfo, option);

            // 上一个选票的删除掉
            if(lastOptions != null){
                for(String op : lastOptions.getOption()){
                    cache.removeSet(Caches.USER_OPTION_SET(topic, op), userInfo);
                }
            }

            // 统计选项人数
            for(String op : option){
                cache.addSet(Caches.USER_OPTION_SET(topic, op), userInfo);
            }

//            if(extraData != null){
//                cache.put(Caches.EXTRA_TYPE(topic), extraData.getClass().getName());
//            }
            cache.addList(Caches.NEW_USER_LIST(topic), uio);

            // 推送到定时器，由定时器自行决定什么时候产生回调。
            topicNotifier.push(topic, uio);
        } catch (Exception ex) {
            log.error(ex.toString(), ex);
        } finally {
            lock.writeUnlock(topic);
        }

    }

    @Override
    public long sum(String topic) {
        lock.readLock(topic);
        try {
            return cache.sizeOfHash(Caches.OPTION_HASH_MAP(topic));
        } finally {
            lock.readUnlock(topic);
        }
    }

    @Override
    public long sumOptions(String topic, String option) {
        lock.readLock(topic);
        try {
            return cache.sizeOfSet(Caches.USER_OPTION_SET(topic, option));
        } finally {
            lock.readUnlock(topic);
        }
    }

    @Override
    public long sumUserInfos(String topic, String userInfo) {
        lock.readLock(topic);
        try {
            UserInfoOption options = cache.getHash(Caches.OPTION_HASH_MAP(topic), userInfo, UserInfoOption.class);
            return options == null ? 0 : options.getOption().length;
        } finally {
            lock.readUnlock(topic);
        }
    }

    @Override
    public String[] options(String topic, String userInfo) {
        lock.readLock(topic);
        try {
            String hashName = topic + "@";
            UserInfoOption options = (UserInfoOption) cache.getHash(hashName, userInfo, UserInfoOption.class);
            if(options == null){
                return new String[0];
            }
            return options.getOption();
        } finally {
            lock.readUnlock(topic);
        }
    }

    @Override
    public Set<String> userInfos(String topic, String option) {
        lock.readLock(topic);
        try {
            // 统计选项人数
            return cache.getSet(Caches.USER_OPTION_SET(topic, option), String.class);
        } finally {
            lock.readUnlock(topic);
        }
    }

    @Override
    public List<UserInfoOption> getAndRefreshNewUserInfo(String topic) {
        lock.readLock(topic);
        try {
            // 新增人数
            List<UserInfoOption> newUserInfos = cache.getList(Caches.NEW_USER_LIST(topic), UserInfoOption.class);
            try {
                cache.cleanSet(topic + "@new");
            } finally {
                return newUserInfos;
            }
        } finally {
            lock.readUnlock(topic);
        }
    }

    @Override
    public void testFlush() {
        topicNotifier.testNotify();
    }

    @Override
    public void flush(String topic) {
        topicNotifier.notify(topic);
    }

    @Override
    public boolean isInit(String topic) {
        return cache.containHash(Caches.OPTION_HASH_MAP(topic));
    }

    @Override
    public void release(String topic) {

        Map<String, UserInfoOption> lastOptionMap = cache.getHash(Caches.OPTION_HASH_MAP(topic), UserInfoOption.class);
        Set<String> options = new HashSet<>();
        for(UserInfoOption ops : lastOptionMap.values()){
            options.addAll(Arrays.asList(ops.getOption()));
        }

        // 上一个选票的删除掉
        if(!isEmpty(options)){
            for(String op : options){
                cache.cleanSet(Caches.USER_OPTION_SET(topic, op));
            }
        }

        cache.cleanHash(Caches.OPTION_HASH_MAP(topic));
        cache.cleanValue(Caches.HOLDER(topic));
        cache.cleanValue(Caches.HOLDER_TYPE(topic));
//        cache.cleanValue(Caches.EXTRA(topic));
//        cache.cleanValue(Caches.EXTRA_TYPE(topic));
        cache.cleanList(Caches.NEW_USER_LIST(topic));


        EasyBus.getInstance().unregisterOptionListener(topic);
    }

    @Override
    public <T> T getHolder(String topic) {
        lock.readLock(topic);
        try {
            String typeName = cache.get(topic + "@type", String.class);
            try {
                return (T) cache.get(topic + "@holder", Class.forName(typeName));
            } catch (ClassNotFoundException e) {
                throw new RuntimeException(e);
            }
        } finally {
            lock.readUnlock(topic);
        }
    }

    @Override
    public Set<UserInfoOption> userInfoOptions(String topic) {
        String hashName = topic + "@";
        Map<String, UserInfoOption> totalOptions = cache.getHash(hashName, UserInfoOption.class);
        return new HashSet<>(totalOptions.values());
    }
}
